use std::{
    collections::HashMap,
    sync::{
        atomic::{AtomicBool, Ordering},
        Arc,
    },
    time::Duration,
};

use crate::{
    model::sc::{
        AddServiceInfo, CommonResponse, CreateServiceInstanceResponse, CreateServiceResponse,
        FindServiceInstanceResponse, InstanceStatus, ServiceInstanceResponse, ServiceSchemaInfo,
    },
    util::{now, now_hw},
    Error, Result,
};
use serde_json::json;
use tokio::{
    sync::{
        mpsc::{channel, Receiver, Sender},
        oneshot,
    },
    time::sleep,
};
use versions::Version;

const PROJECT: &str = "default";
const HEART_BEAT_INTERVAL: u32 = 30;
const HEART_BEAT_RETRY_NUM: u32 = 3;
const CACHE_EXPIRE_DURATION: u32 = 30;

#[derive(Debug, Clone)]
pub struct ServiceCenterOption {
    pub sc_addr: String,
    pub host_name: String,
    pub addr: String,
    pub domain: String,
    pub app_id: String,
    pub service_name: String,
    pub version: String,
}

pub enum Message {
    Add(AddServiceInfo),
    //ServiceName
    Remove(String),
    Call(MessageCall),
    Quit,
}

pub struct MessageCall {
    req: MessageReq,
    tx: oneshot::Sender<MessageResp>,
}

enum MessageReq {
    FindService(String),
}

enum MessageResp {
    Fail(Error),
    FindService(ServiceInstanceResponse),
}

pub struct ServiceCenter {
    tx: Sender<Message>,
    is_quit: Arc<AtomicBool>,
}

impl ServiceCenter {
    pub async fn start(opt: ServiceCenterOption) -> Result<Self> {
        let (tx, rx) = channel(10);
        let mut inner = ServiceCenterInner::new(opt);
        let is_quit = Arc::new(AtomicBool::new(false));
        let is_quit1 = is_quit.clone();
        tokio::spawn(async move {
            info!("service center loop>>>");
            if let Err(e) = inner.do_loop(rx, is_quit1).await {
                error!("do loop failed: {}", e.to_string());
            }
            info!("service center loop<<<");
        });
        Ok(Self { tx, is_quit })
    }

    pub async fn regist_service(&self, info: AddServiceInfo) -> Result<()> {
        self.tx
            .send(Message::Add(info))
            .await
            .map_err(|e| Error::Other(e.to_string()))
    }

    pub async fn find_service(&self, service_name: String) -> Result<ServiceInstanceResponse> {
        match self.call(MessageReq::FindService(service_name), 10).await? {
            MessageResp::FindService(s) => {
                return Ok(s);
            }
            _ => {
                error!("bug! not match! ");
                return Err(Error::Bug);
            }
        }
    }

    async fn call(&self, req: MessageReq, timeout: u32) -> Result<MessageResp> {
        let (tx, mut rx) = oneshot::channel();
        let _ = self
            .tx
            .send(Message::Call(MessageCall { req, tx }))
            .await
            .map_err(|e| Error::Other(e.to_string()))?;
        let s = sleep(Duration::from_secs(timeout.into()));
        tokio::pin!(s);
        tokio::select! {
            _ = &mut s => {
                return Err(Error::Timeout);
            },
            resp = &mut rx => {
                match resp.unwrap() {
                    MessageResp::Fail(e) => {
                        return Err(e);
                    },
                    v => {
                        return Ok(v);
                    }
                }
            },
        }
    }

    fn stop(&self) {
        self.is_quit.store(true, Ordering::SeqCst);
    }
}

impl Drop for ServiceCenter {
    fn drop(&mut self) {
        self.stop();
        info!("drop ServiceCenter");
    }
}
pub struct ServiceCenterInner {
    opt: ServiceCenterOption,
    cache: HashMap<String, CacheInstance>,
}

impl ServiceCenterInner {
    pub fn new(opt: ServiceCenterOption) -> Self {
        Self {
            opt,
            cache: HashMap::new(),
        }
    }

    pub async fn do_loop(
        &mut self,
        mut rx: Receiver<Message>,
        is_quit: Arc<AtomicBool>,
    ) -> Result<()> {
        let mut status = RunStatus::Idle;
        let mut service_id = String::new();
        let mut instance_id = String::new();
        let mut last_hb_ts = now_hw().as_secs();
        info!("start ts {}", last_hb_ts);
        let mut hb_retry_count = 0;
        while !is_quit.load(Ordering::SeqCst) {
            match status {
                RunStatus::Idle => {
                    service_id = match self.regist().await {
                        Ok(id) => id,
                        Err(e) => {
                            error!("regist failed: {}", e.to_string());
                            tokio::time::sleep(Duration::from_secs(5)).await;
                            continue;
                        }
                    };
                    info!("service_id {}", service_id);
                    status = RunStatus::ServiceRegisted;
                    continue;
                }
                RunStatus::ServiceRegisted => {
                    instance_id = match self.regist_instance(&service_id).await {
                        Ok(id) => id,
                        Err(e) => {
                            error!("regist instance failed: {}", e.to_string());
                            tokio::time::sleep(Duration::from_secs(5)).await;
                            status = RunStatus::Idle;
                            continue;
                        }
                    };
                    info!("instance_id {}", instance_id);
                    status = RunStatus::InstanceRegisted;
                    continue;
                }
                RunStatus::InstanceRegisted => {
                    status = RunStatus::Normal;
                    continue;
                }
                RunStatus::Normal => {
                    // check heartbeat
                    let cur = now_hw().as_secs();
                    if cur > last_hb_ts + (HEART_BEAT_INTERVAL / 2) as u64 {
                        info!("send heartbeat: {}", cur);
                        match self.heartbeat(&service_id, &instance_id).await {
                            Ok(_) => {
                                hb_retry_count = 0;
                                last_hb_ts = cur;
                            }
                            Err(e) => {
                                error!("heartbeat failed: {}", e.to_string());
                                hb_retry_count += 1;
                                if hb_retry_count > HEART_BEAT_RETRY_NUM {
                                    error!("retry exceed limit, regitst again!");
                                    status = RunStatus::Idle;
                                    continue;
                                }
                            }
                        }
                    }
                }
            }

            match rx.try_recv() {
                Ok(msg) => match msg {
                    Message::Call(msg) => {
                        info!("handle call");
                        match self.handle_call(&service_id, msg.req).await {
                            Ok(resp) => {
                                info!("call ok, send resp");
                                let _ = msg.tx.send(resp);
                            }
                            Err(e) => {
                                error!("call failed: {}, send resp", e.to_string());
                                let _ = msg.tx.send(MessageResp::Fail(e));
                            }
                        }
                    }
                    Message::Add(info) => {
                        info!("regist schema");
                        if let Err(e) = self.regist_schema(&service_id, &info.schemes).await {
                            error!("regist schema failed: {}", e.to_string());
                            continue;
                        }
                    }
                    Message::Quit => {
                        break;
                    }
                    Message::Remove(_) => {
                        error!("not implement!");
                        continue;
                    }
                },
                Err(_e) => {
                    tokio::time::sleep(Duration::from_millis(10)).await;
                    continue;
                }
            };
        }
        Ok(())
    }

    async fn handle_call(&mut self, service_id: &str, req: MessageReq) -> Result<MessageResp> {
        match req {
            MessageReq::FindService(s) => {
                let cur = now_hw().as_secs();
                if let Some(ref info) = self.cache.get(&s) {
                    if cur < info.ts + CACHE_EXPIRE_DURATION as u64 {
                        info!("service name {} cache hit", s);
                        return Ok(MessageResp::FindService(info.info.clone()));
                    }
                }
                let items = self.find_service_instance(service_id, &s).await?;
                if items.is_empty() {
                    return Err(Error::Empty);
                }
                info!("service name {} instance {}", s, items.len());
                let item = find_best_instance(&items);
                self.cache.insert(
                    s,
                    CacheInstance {
                        ts: cur,
                        info: item.clone(),
                    },
                );
                return Ok(MessageResp::FindService(item));
            }
        }
    }

    pub async fn regist(&self) -> Result<String> {
        let url = format!(
            "http://{}/v4/{}/registry/microservices",
            self.opt.sc_addr, PROJECT
        );
        let client = reqwest::Client::new();
        let res = client
            .post(url)
            .header("x-domain-name", self.opt.domain.as_str())
            .json(&json!({
                "service": {
                    "appId": self.opt.app_id,
                    "serviceName": self.opt.service_name,
                    "version": self.opt.version,
                    "level": "FRONT",
                    "registerBy": "SDK",
                    "framework": {
                        "name": "rust-chassis",
                        "version": env!("CARGO_PKG_VERSION"),
                    },
                    "modTimestamp": now().as_secs().to_string(),
                }
            }))
            .send()
            .await?;
        let resp = res.json::<CreateServiceResponse>().await?;
        Ok(resp.service_id)
    }

    pub async fn regist_instance(&self, service_id: &str) -> Result<String> {
        let url = format!(
            "http://{}/v4/{}/registry/microservices/{}/instances",
            self.opt.sc_addr, PROJECT, service_id
        );
        let client = reqwest::Client::new();
        let ep = format!("rest://{}", self.opt.addr);
        let res = client
            .post(url)
            .header("x-domain-name", self.opt.domain.as_str())
            .json(&json!({
                "instance": {
                    "hostName": self.opt.host_name,
                    "endpoints": [ep],
                    "healthCheck": {
                        "mode": "push",
                        "interval": HEART_BEAT_INTERVAL,
                        "times": 3,
                    },
                    "modTimestamp": now().as_secs().to_string(),
                }
            }))
            .send()
            .await?
            .json::<CreateServiceInstanceResponse>()
            .await?;
        Ok(res.instance_id)
    }

    pub async fn find_service_instance(
        &self,
        consumer_id: &str,
        service_name: &str,
    ) -> Result<Vec<ServiceInstanceResponse>> {
        let url = format!(
            "http://{}/v4/{}/registry/instances?appId={}&serviceName={}&version=latest",
            self.opt.sc_addr, PROJECT, self.opt.app_id, service_name
        );
        let client = reqwest::Client::new();
        let res = client
            .get(url)
            .header("x-domain-name", self.opt.domain.as_str())
            .header("x-consumerid", consumer_id)
            .header("Content-Type", "application/json")
            .send()
            .await?
            .json::<FindServiceInstanceResponse>()
            .await?;
        Ok(res.instances)
    }

    pub async fn regist_schema(
        &self,
        service_id: &str,
        schemas: &Vec<ServiceSchemaInfo>,
    ) -> Result<()> {
        let url = format!(
            "http://{}/v4/{}/registry/microservices/{}/schemas",
            self.opt.sc_addr, PROJECT, service_id
        );
        let client = reqwest::Client::new();
        let res = client
            .post(url)
            .header("x-domain-name", self.opt.domain.as_str())
            .json(&json!({
                "schemas": schemas
            }))
            .send()
            .await?
            .bytes()
            .await?;
        if !res.is_empty() {
            let resp: CommonResponse = serde_json::from_slice(&res)?;
            return Err(resp.into());
        }
        Ok(())
    }

    pub async fn heartbeat(&self, service_id: &str, instance_id: &str) -> Result<()> {
        let url = format!(
            "http://{}/v4/{}/registry/microservices/{}/instances/{}/heartbeat",
            self.opt.sc_addr, PROJECT, service_id, instance_id
        );
        let client = reqwest::Client::new();
        let res = client
            .put(url)
            .header("x-domain-name", self.opt.domain.as_str())
            .send()
            .await?
            .bytes()
            .await?;
        if !res.is_empty() {
            let resp: CommonResponse = serde_json::from_slice(&res)?;
            return Err(resp.into());
        }
        Ok(())
    }

    pub async fn update_status(
        &self,
        service_id: &str,
        instance_id: &str,
        status: InstanceStatus,
    ) -> Result<()> {
        let url = format!(
            "http://{}/v4/{}/registry/microservices/{}/instances/{}/status?value={}",
            self.opt.sc_addr,
            PROJECT,
            service_id,
            instance_id,
            status.to_string(),
        );
        let client = reqwest::Client::new();
        let res = client
            .put(url)
            .header("x-domain-name", self.opt.domain.as_str())
            .send()
            .await?
            .bytes()
            .await?;
        if !res.is_empty() {
            let resp: CommonResponse = serde_json::from_slice(&res)?;
            return Err(resp.into());
        }
        Ok(())
    }

    pub async fn delete_service(&self, service_id: &str) -> Result<()> {
        let url = format!(
            "http://{}/v4/{}/registry/microservices/{}",
            self.opt.sc_addr, PROJECT, service_id
        );
        let client = reqwest::Client::new();
        let res = client
            .delete(url)
            .header("x-domain-name", self.opt.domain.as_str())
            .send()
            .await?
            .bytes()
            .await?;
        if !res.is_empty() {
            let resp: CommonResponse = serde_json::from_slice(&res)?;
            return Err(resp.into());
        }
        Ok(())
    }
}

enum RunStatus {
    Idle,
    ServiceRegisted,
    InstanceRegisted,
    Normal,
}

struct CacheInstance {
    ts: u64,
    info: ServiceInstanceResponse,
}

pub fn find_best_instance(items: &Vec<ServiceInstanceResponse>) -> ServiceInstanceResponse {
    assert!(!items.is_empty());
    let mut index = 0;
    for i in 1..items.len() {
        let item = &items[i];
        if item.endpoints.is_empty() || item.status != "UP" {
            continue;
        }
        if let Some(dst_v) = Version::new(&item.version) {
            if let Some(src_v) = Version::new(&items[index].version) {
                if dst_v > src_v {
                    info!("best version {}", item.version);
                    index = i;
                    continue;
                }
            }
        }
        if item.mod_timestamp > items[index].mod_timestamp {
            index = i;
        }
    }
    info!("best index {}", index);
    items[index].clone()
}
